package com.niit.service

import java.text.SimpleDateFormat

import com.niit.bean.AdClickData
import com.niit.dao.BlackListDao
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream

/*
  需求：将每天对某个广告点击超过100次的用户拉黑，将黑名单数据保存到MySQL当中。

  1.读取Kafka数据之后，并对MySQL中存储黑名单数据做校验；
 2.校验通过则对用户点击广告次数累加一并存入当MySQL当中
 3.在存入MySQL之后数据校验，如果单日超过100次则将该用户拉入黑名单

 Service: 主要是用户分析数据，所以并不是在这里去获得kafka数据
 */
class BlackListService {




  //用于分析数据，所有分析的源数据需要在调用分析方法的时候进行传递，因为我们是实时，所以数据类型应该是DStream
  def dataAnalysis(data:DStream[AdClickData]):Any={
    //   1714960534184 华东 上海 6 5
    //在 Kafka中获取广告点击数据
    val filterData: DStream[((String, String, String), Int)] = filterKafkaData(data)
    checkUserCount(filterData)
  }

  /*
    利用在Kafka中采集出来的流式数据对数据库进行实时检索
    将已经在黑名单的数据过滤掉，过滤后的数据存进每日点击广告次数 数据表中
   */
  def filterKafkaData(data: DStream[AdClickData]): DStream[( (String,String,String) ,Int)  ] = {
    //1.DStream 转换 RDD ,再执行过滤等操作
    data.transform(rdd=>{

       //1.1判断点击用户是否在黑名单当中，如果在黑名单中，就将该数据过滤掉
      val filterRdd =  rdd.filter( line=>{
        //TODO 对黑名单表进行查询，根据userId进行查询
        val blackListDao = new BlackListDao;
        !blackListDao.selectBlackUserById(line.user) // !（都是黑名单中的数据）
      })
      //filterRdd 是没有在黑名单中的用户数据
      //统计每个用户当天点击广告的次数
      val redData: RDD[((String, String, String), Int)] = filterRdd.map(line => {
        val user = line.user //6
        val ad = line.ad // 5
        //   1714960534184 华东 上海 6 5
        // 2024-05-4   华东  上海   张三  加多宝    1
        //时间格式转换
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        val day = sdf.format(new java.util.Date(line.ts.toLong)) // 2024-05-04
        ((day, user, ad), 1) // (   (2024-05-4,张三,加多宝)  ,1 )
      })
      //对相同时间 相同用户 相同广告 的点击次数进行累加
      val reduceData: RDD[((String, String, String), Int)] = redData.reduceByKey(_ + _)

      reduceData //(   (2024-05-4,张三,加多宝)  ,99  )

    })
  }
  /*
  检查用户的点击次数
  如果 不满足拉入黑名单的条件，则存入用户每日点击广告次数表中，当存入用户每日点击广告次数表后发现满足拉黑的条件，则将用户进行拉黑
  一次点击次数大于100 直接拉黑
   */
  private def checkUserCount(ds:DStream[( (String,String,String) ,Int)]):Unit={

    //DStream 没有输出可以执行吗？  print saveAsXXXX foreach
    ds.foreachRDD(rdd=>{
      rdd.foreach{
        case ( (day,user,ad),count)=>{
          println(s"${day},${user},${ad},${count}")
          val blackListDao = new BlackListDao;
          if(count>100){
            //TODO un将用户Id存入黑名单数据表中
            blackListDao.insertBlackList(user)
          }else{
            //TODO 存入用户每日点击广告次数表
            blackListDao.insertUserAdCount(day,user,ad,count)
          }
          //判断更新后的点击数据是否超过 100次，如果超过业拉黑
          //TODO   在数据库中查询该用当日的点击某广告的次数是否超过100次
          val bool: Boolean = blackListDao.checkUserByCount(day, user, ad, 100)
          if(bool){
            blackListDao.insertBlackList(user)
          }
        }
      }
    })

  }
}
